package com.distributed.netty.client;


import com.distributed.netty.entity.Cache;
import com.distributed.netty.entity.ElementJob;
import com.distributed.netty.server.Server;
import com.rcache.core.utils.EmailUtils;
import com.rcache.core.cache.Rcache;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import org.apache.log4j.Logger;

import javax.mail.MessagingException;
import java.io.UnsupportedEncodingException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.concurrent.TimeUnit;

/**
 * @desc 断连重试手柄
 * @author gzy
 * @time 2022-06-05
 */
public class ConnectionWatchdog extends ChannelInboundHandlerAdapter implements TimerTask {
    private final Bootstrap bootstrap;
    private static Timer timer = null;
    private final LinkedList<ElementJob> jobList;
    private volatile  boolean reconnect = true;
    private static int attempts;
    /**缓存*/
    private Rcache cache;
    Logger LOG=Logger.getLogger(ConnectionWatchdog.class);

    public ConnectionWatchdog(Bootstrap bootstrap, Timer timer, boolean reconnect, Rcache cache, LinkedList<ElementJob> jobList) {
        this.bootstrap = bootstrap;
        this.timer = timer;
        this.reconnect = reconnect;
        this.cache=cache;
        this.jobList=jobList;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        LOG.debug("当前链路已经激活，重新尝试连接次数重置为0");
        attempts = 0;
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        LOG.debug("链路关闭");
        LOG.info("slaves："+Cache.set);
        if (reconnect) {
            LOG.debug("链路关闭，进行重连...");
            if (attempts < 1) {
                attempts+=1;
                // 重连的时间间隔越来越长
                int timeout = 2 << attempts;
                LOG.debug("断线重连下次重连时间为:"+timeout+"秒");
                timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);

            }else {
                EventLoopGroup work = bootstrap.group();
                work.shutdownGracefully();
                timer.stop();
                HashSet<String> slaves = Cache.set;

                String ip = "";
                int max=0;
                //遍历选举
                for (String slaveIp:slaves) {
                    String[] strs = slaveIp.split(":");
                    String address = strs[0].replace("/","");
                    int ipPort = Integer.parseInt(strs[1]);
                    if (max<ipPort){
                        ip=address;
                        max=ipPort;
                    }
                }




                //获取本机ip
                InetAddress inet4Address = Inet4Address.getLocalHost();
                String ipSelf = inet4Address.toString().split("/")[1];
                LOG.info("当选的ip=："+ip+", 本机ip="+ipSelf);
                if (ip.equals(ipSelf)||ip.equals("127.0.0.1")){
                    LOG.debug("本机当选，启动服务中...");
                    Server server=new Server(cache);
                    server.start();
                    //如果开启了预警系统则异步发送预警信息
                    if (cache.isWarning()){
                        new Thread(new Runnable() {
                            @Override
                            public void run() {
                                //获取本机ip
                                try {
                                    //发送告警信息
                                    EmailUtils.createMineMessage("slave:"+ipSelf+"的预警信息","原master服务端宕机断线，该slave当选master:"+ipSelf+"，请尽快排查master宕机原因",cache.getManagerEmail());
                                } catch (UnsupportedEncodingException | MessagingException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        }).start();
                    }
                }else{
                    LOG.debug("本机落选，正在连接...");
                    //如果开启了预警系统则异步发送预警信息
                    if (cache.isWarning()){
                        String finalIp = ip;
                        new Thread(new Runnable() {
                            @Override
                            public void run() {
                                //获取本机ip
                                try {
                                    //发送告警信息
                                    EmailUtils.createMineMessage("slave:"+ipSelf+"的预警信息","原master服务端宕机断线，ip为："+ finalIp +"的slave当选master，请尽快排查master宕机原因",cache.getManagerEmail());
                                } catch (UnsupportedEncodingException | MessagingException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        }).start();
                    }
                    //休眠2秒等待当选机器启动服务
                    Thread.sleep(2000);
                    Client client=new Client(cache,ip);
                    client.start();
                }
            }
        }


        ctx.fireChannelInactive();
    }


    @Override
    public void run(Timeout timeout) throws Exception {
        ChannelFuture future;
        // bootstrap 已经初始化结束，只需将handler传入即可
        synchronized (bootstrap) {
            bootstrap.handler(new HeartBeatClientInitializer(bootstrap, timer, reconnect,cache, jobList));
            future = bootstrap.connect(cache.getDistributeIP(), cache.getDistributePort());
        }
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) throws Exception {
                boolean succeed = f.isSuccess();
                if (!succeed) { // 如果重连失败，再次发出重连事件，最多尝试12次
                    LOG.debug("重连失败");
                    f.channel().pipeline().fireChannelInactive();
                } else {
                    LOG.debug("重连成功");
                }
            }
        });

        LOG.debug("timeout task : " + attempts);
    }

}